package com.hzh.flink.kafka

import org.apache.kafka.clients.consumer.{ConsumerRecord, ConsumerRecords, KafkaConsumer}
import java.{lang, util}
import java.time.Duration
import java.util.Properties

object Demo3KafkaConsumer {
  def main(args: Array[String]): Unit = {

    /**
     * 1、创建消费者
     *
     */

    val properties = new Properties()

    properties.setProperty("bootstrap.servers", "master:9092,node2:9092,node2:9092")

    //key 和value 反序列化的类
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")


    /**
     * earliest
     * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
     * latest  默认
     * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产认值生的该分区下的数据
     * none
     * topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     *
     */

    properties.setProperty("auto.offset.reset", "earliest")


    //消费者组
    properties.setProperty("group.id", "asdasd")


    val consumer = new KafkaConsumer[String, String](properties)

    /**
     * 2、订阅一个topic,可以一次订阅多个topic
     *
     */

    val topics = new util.ArrayList[String]()
    topics.add("student")
    consumer.subscribe(topics)


    while (true) {
      println("正在消费")

      /**
       * 消费数据, 需要这一个超时时间
       *
       */
      val consumerRecords: ConsumerRecords[String, String] = consumer
        .poll(Duration.ofSeconds(2))

      //解析数据
      val records: lang.Iterable[ConsumerRecord[String, String]] = consumerRecords
        .records("student")

      val iterRecord: util.Iterator[ConsumerRecord[String, String]] = records.iterator()

      while (iterRecord.hasNext) {
        //获取一行数据
        val record: ConsumerRecord[String, String] = iterRecord.next()

        val topic: String = record.topic() //topic
        val offset: Long = record.offset() //数据偏移量
        val key: String = record.key() //数据的key ,默认没有指定的情况下时null
        val value: String = record.value() //保存的数据
        val ts: Long = record.timestamp() //时间戳，默认时存入的时间

        println(s"$topic\t$offset\t$key\t$value\t$ts")
      }


    }

    //关闭链接
    consumer.close()

  }
}
